package com.ctbri.vSearch.dao.es;


import com.ctbri.vSearch.bean.dto.ArticleAndTermScore;
import com.ctbri.vSearch.bean.dto.SearchItem;
import com.ctbri.vSearch.bean.po.ArticleItem;
import com.ctbri.vSearch.bean.po.GzItem;
import com.ctbri.vSearch.bean.proto.ArticleItemProto;
import com.ctbri.vSearch.dao.es.multiThread.ESSearchDao;
import com.ctbri.vSearch.dao.es.multiThread.EsQueryThread;
import com.ctbri.vSearch.frame.RedisPooledConnection;
import com.ctbri.vSearch.util.CountScore;
import com.ctbri.vSearch.util.TimeUtil;

import java.io.ByteArrayInputStream;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.Resource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Repository;
import redis.clients.jedis.Jedis;

/**
 * Created by Sang on 2/20/17.
 */
@Repository
public class ScoreModel {
	
	@Autowired
    private TaskExecutor taskExecutor;
	@Resource
	private ESSearchDao articleItemSearchDao;
	
	/*//存储所有的查询结果
    private Map<ArticleItem,Double> articleScore = new ConcurrentHashMap<ArticleItem, Double>();
    //存储所有的查询结果
    private Map<ArticleItem, HashMap<String, Float>> termScore = new ConcurrentHashMap<ArticleItem, HashMap<String,Float>>();
    */
   /* //声明比较器并用TreeMap排序
    private ValueComparator vc = new ValueComparator(articleScore);
    private TreeMap<ArticleItem,Double> treeMap = new TreeMap<ArticleItem, Double>(vc);
    */
	
    public void scoreModel(Map<String,Double> internalResults, ArticleAndTermScore articleAndTermScore ,RedisPooledConnection rpc){//wk
    	//存储所有的查询结果
        Map<ArticleItem,Double> articleScore = new ConcurrentHashMap<ArticleItem, Double>();
        //存储所有的查询结果
        Map<ArticleItem, HashMap<String, Float>> termScore = new ConcurrentHashMap<ArticleItem, HashMap<String,Float>>();
        
    	//声明比较器并用TreeMap排序
        ValueComparator vc = new ValueComparator(articleScore);
        TreeMap<ArticleItem,Double> treeMap = new TreeMap<ArticleItem, Double>(vc);

    	
   		ConcurrentHashMap<Integer, List<ArticleItem>> results = new ConcurrentHashMap<>();
   		String key ="articles_spider";//es的查询范围
   		Set<String> words =internalResults.keySet();
		CountDownLatch countDownLatch = new CountDownLatch(words.size());
        //为了避免多个线程抢一个redis连接，这里降低了性能，一次只允许一个线程启动，待优化
     //   CountDownLatch countDownLatch = new CountDownLatch(1);
		AtomicInteger atomcCount = new AtomicInteger(0);//线程安全自增
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
		for(String word: words){
            Jedis client = rpc.getClient();
            System.out.println("目前为："+word);
            Double weight = internalResults.get(word);
            //针对每个word判断是否在redis缓存中
            //命名规则：是否在缓存中的标志：word+_ES
            // 每个word的ES查询结果：word+_article_ES
            TimeUtil timeutil = new TimeUtil();
            long start = timeutil.startTime();
            if(client.exists(word+"_ES")) {
      //          client.del(word + "_ES"); //test
                Iterator<String> iterNode=client.hkeys(word+"_article_ES").iterator();
                //将redis缓存中的结果取出，填充
                while (iterNode.hasNext()) {
                    String key2 = iterNode.next();
                    ByteArrayInputStream input = null;
                    ArticleItemProto.ArticleItem articleItemProto = null;
                    try {
                        input = new ByteArrayInputStream(
                                client.hmget(word + "_article_ES", key2).get(0)
                                        .getBytes("ISO-8859-1"));
                        articleItemProto = ArticleItemProto.ArticleItem.parseFrom(input);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    ArticleItem ai = new ArticleItem();
                    ai.setId(articleItemProto.getId());
                    ai.setTitle(articleItemProto.getTitle());
                    ai.setContent(articleItemProto.getContent());
                    ai.setOffcial_account(articleItemProto.getOffcialAccount());
                    String dat = articleItemProto.getDate();
                    if (null == dat || dat =="") {
                        dat = articleItemProto.getTime();
                    }
                    ai.setDate(dat);
                    ai.setUrl(articleItemProto.getUrl());
                    String score = articleItemProto.getScore();
                    Date date = null;
                    try {
                        date = dateFormat.parse(dat);
                    } catch (ParseException e) {
                        e.printStackTrace();
                    }
                    //将得分写入articleScore与termScore中
                    CountScore.countScore(date, articleScore, termScore, ai,
                            Float.parseFloat(score),
                            word, weight);
                }
                long end = timeutil.stopTime();
                System.out.println("从redis缓存中查询ES信息 "+word+" 时间："+(end - start));
                System.out.println("从redis缓存查询ES的 "+word+" 完毕");
                countDownLatch.countDown();


            }
            //否则，视为第一次查询ES数据库，从数据库中查询，并加入redis中
            else
            {
//                new Thread(
//                        new EsQueryThread(atomcCount,articleItemSearchDao,key,word,weight,countDownLatch,results,
//                                articleScore,termScore, client, rpc)).start();
                taskExecutor.execute(
                        new EsQueryThread(atomcCount,articleItemSearchDao,key,word,weight,countDownLatch,results,
                                articleScore,termScore,client,rpc));

//                    new EsQueryThread(atomcCount,articleItemSearchDao,key,word,weight,countDownLatch,results,
//                               articleScore,termScore, client).run(); //这里没有用到多线程，待优化
//                long end = timeutil.stopTime();
//                System.out.println("从ES中查询 "+word+" 时间："+(end - start));

            }
		}
        try {
            System.out.println("等待ES线程结束");
            countDownLatch.await();//在此等待所有线程执行完毕
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        treeMap.putAll(articleScore);
        
        //接收结果
        articleAndTermScore.setTreeMap(treeMap);
        articleAndTermScore.setTermScore(termScore); 
    }


    class ValueComparator implements Comparator<ArticleItem> {
        Map<ArticleItem, Double> base;

        public ValueComparator(Map<ArticleItem, Double> base) {
            this.base = base;
        }
      
        public int compare(ArticleItem a, ArticleItem b) {
            if (base.get(a) >= base.get(b)) {
                return -1;
            } else {
                return 1;
            } // returning 0 would merge keys
        }
    }
}

